/**
 * File: ParallelController.java
 * Created by: mhaimel
 * Created on: 24 Jun 2009
 * CVS:  $Id: ParallelController.java,v 1.2 2009/10/09 14:45:20 mhaimel Exp $
 */
package uk.ac.ebi.curtain.controller;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;

import uk.ac.ebi.curtain.CurtainContext;
import uk.ac.ebi.curtain.configuration.CurtainConfig;
import uk.ac.ebi.curtain.management.IController;
import uk.ac.ebi.curtain.management.IControllerReport;
import uk.ac.ebi.curtain.management.IInterceptor;
import uk.ac.ebi.curtain.management.IProcessor;
import uk.ac.ebi.curtain.management.InterceptorInfo;
import uk.ac.ebi.velvet.VelvetUncheckedException;
import uk.ac.ebi.velvet.exec.ProcessorService;
import uk.ac.ebi.velvet.exec.impl.AbstractJobQueue;

import com.google.inject.Inject;

/**
 * @author mhaimel
 *
 */
public abstract class ParallelController<I extends Serializable,O extends Serializable,P extends IProcessor<I, O>> extends AbstractJobQueue<O, P,IControllerReport> implements IController<I, O> {
	
	private String queueName;
	private Integer parallel;
	private List<IInterceptor<O>> interceptorList;

	private CurtainContext context;
	private CurtainConfig config;

	private ProcessorService processorService;
	
	public ParallelController() {
		this.interceptorList = new ArrayList<IInterceptor<O>>();
	}	

	public void setQueueName(String queueName) {
		if(null == this.queueName){
			this.queueName = queueName;
		} else {
			throw new VelvetUncheckedException(
					"Queue name already set to " + this.queueName + "! New name " + queueName + " Ignored");
		}
	}

	public void setParallel(Integer parallel) {
		if(null == this.parallel){
			this.parallel = parallel;
		} else {
			throw new VelvetUncheckedException(
					"Queue parallel already set to " + this.parallel + "! New value " + parallel + " Ignored");
		}
	}

	@Override
	protected void preCheck() {
		this.processorService.registerCategory(getQueueName(), getQueueParallel());
		if(null == getProcessorService()){
			throw new VelvetUncheckedException("No Processor Service registered!!!");
		}
		if(null == getInterceptors()){
			throw new VelvetUncheckedException("No Interceptor registered!!!");
		}
		super.preCheck();
	}
	
	@Override
	protected int sleepTime() {
		return 300;
	}
	
	/**
	 * {@inheritDoc}
	 */
	@Override
	protected void doShutdown() {
		getLog().info("Finished processing jobs for " + getQueueName() + "!");
		this.getProcessorService().shutdown(getQueueName());
		interceptorsShutdown(false, buildInterceptorInfo());
	}

	private void interceptorsShutdown(boolean force, InterceptorInfo info) {
		for(IInterceptor<O> inter : getInterceptors()){
			inter.shutDown(force, info);
		}
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	protected void forceShutdown() {
		this.getProcessorService().shutdown(getQueueName(), true);
		interceptorsShutdown(true, buildInterceptorInfo());
	}

	@Override
	protected Future<O> submitJob(P job) {
		return getProcessorService().process(getQueueName(), job);
	}
	
	protected String getQueueName() {
		return queueName;
	}
	
	protected List<IInterceptor<O>> getInterceptors() {
		return interceptorList;
	}

	public void addInterceptor(IInterceptor<O> interceptor) {
		this.interceptorList.add(interceptor);
	}
	
	@Inject
	public void setProcessorService(ProcessorService srv) {
		if(null != this.processorService){
			throw new VelvetUncheckedException("Processor Service already available!!!");
		}
		if(null == this.getQueueName()){
			throw new VelvetUncheckedException("Queue name missing!");
		}
		if(null == this.getQueueParallel()){
			throw new VelvetUncheckedException("Parallel for Queue >" + this.getQueueName() +  "< is missing!");
		}
		this.processorService = srv;
	}

	protected Integer getQueueParallel() {
		return parallel;
	}

	protected ProcessorService getProcessorService() {
		return this.processorService;
	}
	
	@Override
	protected void processResult(Future<O> res, P job) {
		O result = extractResult(res, job);
		if(null != result){
			interceptorHandle(result);
		}
	}
	
	protected void interceptorHandle(O result) {
		for(IInterceptor<O> inter : getInterceptors()){
			inter.handle(result, buildInterceptorInfo());
		}
	}

	protected InterceptorInfo buildInterceptorInfo() {
		return new InterceptorInfo(getQueueName());
	}

	protected O extractResult(Future<O> res, P job) {
		try {
			O out = res.get();
			return out;
		} catch (Exception e) {
			throw new VelvetUncheckedException("Problems with Job " + job,e);
		}		
	}

	public void handle(I input, InterceptorInfo info) {
		P job = buildJob(input,info);
		if(null != job){
			job.setInput(input);
			submit(job);
		}
	}
	
	protected abstract P buildJob(I input, InterceptorInfo info);
	
	public void shutDown(boolean force, InterceptorInfo info) {
		if(!force){
			getLog().info("All jobs received for " + getQueueName());
		}
		super.shutdown(force);
	}

	@Inject
	public void setContext(CurtainContext context) {
		this.context = context;
	}

	protected CurtainContext getContext() {
		return context;
	}	

	@Inject
	public void setConfig(CurtainConfig config) {
		this.config = config;
	}
	public CurtainConfig getConfig() {
		return config;
	}

}
